package org.example.sink;

import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.example.flink.pojo.WaterSensor;
import org.example.function.WaterSensorFunction;

import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @Author: tang
 * @Description:
 * @Date 2025/2/19 10:55
 */
public class MysqlFink {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> streamSource = environment.socketTextStream("127.0.0.1", 7777);

        SinkFunction<WaterSensor> sink = JdbcSink.sink("insert into ws values(?,?,?)",
                (JdbcStatementBuilder<WaterSensor>) (preparedStatement, waterSensor) -> {
                    preparedStatement.setObject(1, waterSensor.id);
                    preparedStatement.setObject(2, waterSensor.vt);
                    preparedStatement.setObject(3, waterSensor.vc);
                    System.out.println("写入数据：" + waterSensor);
                }, new JdbcExecutionOptions.Builder()
                        .withMaxRetries(3)
                        // 3秒发送，，或者批发送
                        .withBatchSize(100)
                        .withBatchIntervalMs(3000)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withConnectionCheckTimeoutSeconds(60)
                        .withUsername("root")
                        .withPassword("7210Tang")
                        .withDriverName("com.mysql.cj.jdbc.Driver")
                        .withUrl("jdbc:mysql://127.0.0.1:3306/test?serverTimezone=Asia/Shanghai&useUnicode=true&characterEncoding=UTF-8")
                        .build()
        );

        streamSource
                .map(new WaterSensorFunction())
                .addSink(sink);

        environment.execute();

    }
}
